home *** CD-ROM | disk | FTP | other *** search
- # Source Generated with Decompyle++
- # File: in.pyo (Python 2.4)
-
- from test.test_support import verbose
- import random
- import dummy_threading as _threading
- import time
-
- class TestThread(_threading.Thread):
-
- def run(self):
- global running, running
- delay = 0
- if verbose:
- print 'task', self.getName(), 'will run for', delay, 'sec'
-
- sema.acquire()
- mutex.acquire()
- running = running + 1
- if verbose:
- print running, 'tasks are running'
-
- mutex.release()
- time.sleep(delay)
- if verbose:
- print 'task', self.getName(), 'done'
-
- mutex.acquire()
- running = running - 1
- if verbose:
- print self.getName(), 'is finished.', running, 'tasks are running'
-
- mutex.release()
- sema.release()
-
-
-
- def starttasks():
- for i in range(numtasks):
- t = TestThread(name = '<thread %d>' % i)
- threads.append(t)
- t.start()
-
-
-
- def test_main():
- global numtasks, sema, mutex, running, threads
- numtasks = 10
- sema = _threading.BoundedSemaphore(value = 3)
- mutex = _threading.RLock()
- running = 0
- threads = []
- starttasks()
- if verbose:
- print 'waiting for all tasks to complete'
-
- for t in threads:
- t.join()
-
- if verbose:
- print 'all tasks done'
-
-
- if __name__ == '__main__':
- test_main()
-
-